package com.project.capture5.app;

import com.project.capture5.bean.UserBehavior;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author Shelly An
 * @create 2020/9/18 15:19
 */
public class PageView {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //1. 从文件读取数据，转换成bean对象
        SingleOutputStreamOperator<UserBehavior> userBehaviorDS = env.readTextFile("Data/UserBehavior.csv")
                .map(new MapFunction<String, UserBehavior>() {
                    @Override
                    public UserBehavior map(String value) throws Exception {
                        String[] datas = value.split(",");

                        return new UserBehavior(Long.valueOf(datas[0]),
                                Long.valueOf(datas[1]),
                                Integer.valueOf(datas[2]),
                                datas[3],
                                Long.valueOf(datas[4]));
                    }
                });


        //2. 参考WordCount思路，实现pv的统计
        //2.1 过滤出pv行为
        SingleOutputStreamOperator<UserBehavior> userBehaviorFilter = userBehaviorDS.filter(
                (FilterFunction<UserBehavior>) value -> "pv".equals(value.getBehavior()));

        //2.2 只关心pv行为 转换成二元组(pv,1) 所以第一个元素写死了，省空间
        SingleOutputStreamOperator<Tuple2<String, Integer>> pvAndOneTuple2 = userBehaviorFilter.map(new MapFunction<UserBehavior, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(UserBehavior value) throws Exception {
                return Tuple2.of("pv", 1);
            }
        });


        //2.3 按照第一个位置的元素 分组
        KeyedStream<Tuple2<String, Integer>, Tuple> pvAndOneKS = pvAndOneTuple2.keyBy(0);

        //普通的流不能做聚合，只有分组的流才能做聚合
        //聚合算子智能在分组后调用，也就是在keyedStream才能调用sum
        SingleOutputStreamOperator<Tuple2<String, Integer>> pvDS = pvAndOneKS.sum(1);

        //2.4 求和  434349
        pvDS.print("ps");

        env.execute();
    }
}
